package com.zz.common.kafka

import akka.actor.ActorSystem
import akka.stream.Materializer
import akka.stream.scaladsl._
import cats.effect.IO
import com.typesafe.config.ConfigFactory
import org.apache.kafka.common.serialization.{ByteArraySerializer, Serializer, StringSerializer}
import scala.concurrent.Future

import com.zz.common.GlobalContext
import com.zz.dsp.idl.DspEvent


object KafkaPipeline {

  def apply()(implicit ctx: GlobalContext[IO]): KafkaPipeline = new KafkaPipeline {

    implicit val actorSystem: ActorSystem = ctx.system

    implicit val mt: Materializer = ctx.mat

    private[this] val config = ConfigFactory.load()

    val kafkaFull1Topic: String = config.getString("kafka.dsp-id-mapping.topic-in")

  }

}

trait KafkaPipeline extends ProducerPipeLine[String, Array[Byte], DspEvent] {

  implicit def actorSystem: ActorSystem

  implicit def mt: Materializer

  def kafkaFull1Topic: String

  override val producerKeySerializer: Serializer[String] = new StringSerializer

  override val producerValueSerializer: Serializer[Array[Byte]] = new ByteArraySerializer

  override def keyTransfer(t: DspEvent): String = t.requestId

  override def valueTransfer(t: DspEvent): Array[Byte] = t.toByteArray

  lazy val kafkaFull1Queue: SourceQueueWithComplete[DspEvent] = queue(kafkaFull1Topic)


  def toKafka(log: DspEvent): Future[_] = {
    kafkaFull1Queue.offer(log)
  }

}
